package cn.openea.perpartner.common.rabbit.listener;

import cn.openea.perpartner.common.rabbit.task.AsyncTask;
import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author Weird077
 * @description TODO
 * @date 2022-06-01 23:06
 */
@Component
public class AsyncListener {
    Logger logger = LoggerFactory.getLogger(AsyncListener.class);
    @Autowired
    AmqpTemplate amqpTemplate;
    @Autowired
    AsyncTask asyncTask;
    @RabbitListener(queues = "${rabbitmq.basic.queue}")
    public void doInvoke(JSONObject object, @Header(AmqpHeaders.REPLY_TO) String replyTo, @Header(AmqpHeaders.CORRELATION_ID) String correlationId) throws Exception {
        logger.info("use a fixed reply queue {}, it is necessary to provide correlation data {} so that replies can be correlated to requests", replyTo,correlationId);
        ListenableFuture<Object> asyncResult = asyncTask.asyncInvoke(object);
        ListenableFutureCallback<Object> callback = new ListenableFutureCallback<Object>() {
            @Override
            public void onSuccess(Object result) {
                amqpTemplate.convertAndSend(replyTo, result, message -> {
                    message.getMessageProperties().setCorrelationId(correlationId);
                    return message;
                });
            }
            @Override
            public void onFailure(Throwable ex) {
                ex.printStackTrace();
            }
        };
        asyncResult.addCallback(callback);
    }
}
